package com.atguigu.app

import java.text.SimpleDateFormat
import java.util
import java.util.Date

import com.alibaba.fastjson.JSON
import com.atguigu.bean.{CouponAlertInfo, EventLog}
import com.atguigu.constants.GmallConstants
import com.atguigu.utils.{MyEsUtil, MyKafkaUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.{Minutes, Seconds, StreamingContext}

import scala.util.control.Breaks._

object AlertApp {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf
    val sparkConf: SparkConf = new SparkConf().setAppName("AlertApp").setMaster("local[*]")

    //2.创建StreamingContext
    val ssc: StreamingContext = new StreamingContext(sparkConf, Seconds(5))

    //3.消费kafka数据
    val kafkaDStream: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(GmallConstants.KAFKA_TOPIC_EVENT, ssc)

    //4.将数据转为样例类
    val sdf: SimpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH")
    val EventLogDStream = kafkaDStream.mapPartitions(partition => {
      partition.map(record => {
        val eventLog: EventLog = JSON.parseObject(record.value(), classOf[EventLog])
        val times: String = sdf.format(new Date(eventLog.ts))

        eventLog.logDate = times.split(" ")(0)
        eventLog.logHour = times.split(" ")(1)
        (eventLog.mid, eventLog)
      })
    })

    //5.开启一个五分钟的窗口
    val windowDStream = EventLogDStream.window(Minutes(5))

    //6.对相同mid的数据进行聚合
    val midToIterEventLogDStream: DStream[(String, Iterable[EventLog])] = windowDStream.groupByKey()


    //7.根据条件进行过滤
    /**
      * 三次及以上用不同账号登录并领取优惠劵，（1.获取到涉及领取优惠券行为的数据2.对账号进行去重3.计       算不同账号的个数）
      * 并且过程中没有浏览商品（判断用户除了领优惠券外是否浏览商品）。
      */
    val boolToCouponDStream: DStream[(Boolean, CouponAlertInfo)] = midToIterEventLogDStream.mapPartitions(partition => {
      partition.map { case (mid, iter) =>
        //创建set集合用来存放领优惠券但是没有浏览商品的用户id
        val uids: util.HashSet[String] = new util.HashSet[String]()
        //创建set集合用来存放领优惠券所涉及的商品
        val itemids: util.HashSet[String] = new util.HashSet[String]()
        //创建List集合用来存放用户所涉及的事件
        val events: util.ArrayList[String] = new util.ArrayList[String]()

        //创建一个标志位，用来判断用户是否浏览商品
        var bool: Boolean = true

        //a.首先遍历出迭代器中的每条数据
        breakable {
          iter.foreach(log => {
            //将用户涉及的行为添加到list集合中
            events.add(log.evid)
            //b.判断用户是否有浏览商品行为
            if ("clickItem".equals(log.evid)) {
              bool = false
              //跳出循环
              break
            } else if ("coupon".equals(log.evid)) {
              //c.用户没有浏览商品,但是领优惠券
              //将用户id存放到集合中
              uids.add(log.uid)
              itemids.add(log.itemid)
            }
          })
        }
        //生成疑似预警日志
        (uids.size() >= 3 && bool, CouponAlertInfo(mid, uids, itemids, events, System.currentTimeMillis()))
      }
    })

    //8.生成预警日志
    val couponAlertInfoDStream: DStream[CouponAlertInfo] = boolToCouponDStream.filter(_._1).map(_._2)

    couponAlertInfoDStream.print()

    //9.将数据写入ES
    couponAlertInfoDStream.foreachRDD(rdd=>{
      rdd.foreachPartition(partition=>{
        val list: List[(String, CouponAlertInfo)] = partition.toList.map(info => {
          //将数据转为list集合并且转为k，v类型的，k是docid，v是具体的数据
          (info.mid + info.ts / 1000 / 60, info)
        })
        MyEsUtil.insertBulk(GmallConstants.ES_INDEX_ALERT+"210927",list)
      })
    })

    //10.启动任务并阻塞
    ssc.start()
    ssc.awaitTermination()
  }

}
